package com.gitee.dengmin.logback;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * @Author dengmin
 * @Created 2020/6/8 下午5:10
 */
public class KafkaProduceClient {
    private static final long serialVersionUID = 1L;
    private static KafkaProduceClient instance;

    private KafkaProducePool pool;

    public static KafkaProduceClient getInstance(String hosts){
        if(null == instance){
            synchronized (KafkaProduceClient.class){
                if(null == instance){
                    instance = new KafkaProduceClient(hosts);
                }
            }
        }
        return instance;
    }

    private KafkaProduceClient(String hosts){
        this.pool = new KafkaProducePool(hosts);
    }

    public void push(String topic, String message) throws Exception {
        KafkaProducer<String,String> producer = pool.get();
        ProducerRecord<String,String> record = new ProducerRecord<String, String>(topic, message);
        producer.send(record);
        pool.returnObject(producer);
    }

}
